package com.haiyou.dc.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

/**
 * @author root
 */
@Configuration
@Slf4j
public class RabbitmqConfig {

    @Autowired
    private Environment env;
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * 单一消费者
     *
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        log.debug("--------------------------------->singleListenerContainer配置");
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        return factory;
    }

    /**
     * 多个消费者
     *
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        log.debug("--------------------------------->multiListenerContainer配置");
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.concurrency", int.class));
        factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.max-concurrency", int.class));
        factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.simple.prefetch", int.class));
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
            }
        });
        return rabbitTemplate;
    }


    /**
     * 创建交换机
     *
     * @return
     */
    @Bean
    public DirectExchange basicDirectExchange() {
        return new DirectExchange(env.getProperty("mq.basic.info.exchange"), true, false);
    }

    /**
     * 创建队列
     *
     * @return
     */
    @Bean
    public Queue basicQueue() {
        return new Queue(env.getProperty("mq.basic.info.queue"));
    }

    /**
     * 创建绑定
     *
     * @return
     */
    @Bean
    public Binding basicBinding() {
        return BindingBuilder.bind(basicQueue()).to(basicDirectExchange()).with(env.getProperty("mq.basic.info.routeKey"));
    }

    @Bean
    public DirectExchange productExchange() {
        return new DirectExchange(env.getProperty("mq.basic.product.exchange"), true, false);
    }

    @Bean
    public Queue productQueue() {
        return new Queue(env.getProperty("mq.basic.product.queue"), true, false, false, null);
    }

    @Bean
    public Binding productQueueBindExchange() {
        return BindingBuilder.bind(productQueue()).to(productExchange()).with(env.getProperty("mq.basic.product.routeKey"));
    }

    /**
     * 日志消息模型
     *
     * @return
     */
    @Bean
    public TopicExchange systemLogExchange() {
        return new TopicExchange(env.getProperty("mq.log.exchange"), true, false);
    }

    @Bean
    public Queue systemLogQueue() {
        return new Queue(env.getProperty("mq.log.queue"), true, false, false, null);
    }

    @Bean
    public Binding systemLogBinding() {
        return BindingBuilder.bind(systemLogQueue()).to(systemLogExchange()).with(env.getProperty("mq.log.routeKey"));
    }
}